package com.yiquan.search.service;

import com.mongodb.spark.MongoSpark;
import com.mongodb.spark.rdd.api.java.JavaMongoRDD;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.mllib.recommendation.ALS;
import org.apache.spark.mllib.recommendation.MatrixFactorizationModel;
import org.apache.spark.mllib.recommendation.Rating;
import org.apache.spark.sql.SparkSession;
import org.bson.Document;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import scala.Tuple2;

import javax.naming.ldap.ManageReferralControl;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * sparkMLlib推荐系统(离线)
 *
 * @author Tyrone
 * @date 2022/3/14 19:51
 */
@Service
public class SparkRecommend {

    @Autowired
    private RedisTemplate redisTemplate;

    public static final String RECOMMEND_VIDEO_PREFIX = "RECOMMEND_VIDEO";

    @Scheduled(cron = "0 0 0 * * *")
    public void startRecommend(Integer recommendNums) {
        SparkSession sparkSession = SparkSession.builder()
                .appName("LogAnalyse")
                .master("local[*]")
                .config("spark.mongodb.input.uri", "mongodb://192.168.255.138:27017/yiquan_recommend.browseAction")
                .getOrCreate();

        JavaSparkContext jsc = new JavaSparkContext(sparkSession.sparkContext());
        jsc.setLogLevel("INFO");

        /**
         * 加载MongoDB中的数据
         */
        JavaMongoRDD<Document> rdd = MongoSpark.load(jsc);
        rdd.foreach(document -> System.out.println(document.toJson()));

        /**
         * 同一个用户对不同的视频可能存在相同的操作  合并操作 操作的分数也需要累加
         */
        JavaRDD<Document> values = rdd.mapToPair(document -> {
            Long userId = document.getLong("userId");
            Long videoId = document.getLong("videoId");
            return new Tuple2<>(userId + "_" + videoId, document);
        }).reduceByKey((v1, v2) -> {
            double newScore = v1.getDouble("score") + v2.getDouble("score");
            v1.put("score", newScore);
            return v1;
        }).values();

        List<Long> userIdList = rdd.map(v1 -> v1.getLong("userId")).distinct().collect();

        /**
         * 为了将数据集合进行切割，60%用户训练 20%用于校验 20%用于测试
         * 将JavaRDD转成JavaPairRDD 后面根据key去划分到各个集合
         * key：时间戳模10 value：Rating
         */
        JavaPairRDD<Long, Rating> ratings = values.mapToPair(document -> {
            Long date = document.getLong("browseDate");
            int userId = document.getLong("userId").intValue();
            int videoId = document.getLong("videoId").intValue();
            Double score = document.getDouble("score");
            Rating rating = new Rating(userId, videoId, score);
            return new Tuple2<>(date % 10, rating);
        });

        ratings.foreach(longRatingTuple2 -> System.out.println(longRatingTuple2._1() + ":" + longRatingTuple2._2()));

        /**
         * 根据数据集训练校验测试后，获取最佳模型
         */
        SparkRecommend sparkRecommend = new SparkRecommend();
        MatrixFactorizationModel bestModel = sparkRecommend.bestModel(ratings);

        /**
         * 模型开始推荐
         */

        for (Long userId : userIdList) {
            Rating[] recommendProducts = bestModel.recommendProducts(userId.intValue(), recommendNums);

            List<Integer> videoIdList = new ArrayList<>();

            for (Rating product : recommendProducts) {
                videoIdList.add(product.product());
            }

            pushRecommend(userId, videoIdList);

        }

        jsc.close();
    }

    /**
     * 将模型能够推荐的视频放进Redis中 （过期时间为1天）
     *
     * @param userId
     * @param videoIdList
     */
    public void pushRecommend(Long userId, List<Integer> videoIdList) {
        redisTemplate.opsForValue().set(RECOMMEND_VIDEO_PREFIX + userId, videoIdList, 1, TimeUnit.DAYS);
    }

    /**
     * 计算最佳模型
     *
     * @param ratings
     * @return
     */
    public MatrixFactorizationModel bestModel(JavaPairRDD<Long, Rating> ratings) {
        long numRating = ratings.count();
        long numUsers = ratings.map(v -> ((Rating) v._2()).user()).distinct().count();
        long numMovies = ratings.map(v -> ((Rating) v._2()).product()).distinct().count();
        System.out.println("用户： " + numUsers + " 动态：" + numMovies + "评论：" + numRating);

        /**
         * 60%用于训练20%用于检测 20%用于测试
         */
        Integer numPartitions = 4;

        /**
         * 训练集
         */
        JavaRDD<Rating> training = ratings.filter(v -> v._1() < 6)
                .values()
                .repartition(numPartitions)
                .cache();

        /**
         * 校验集合
         */
        JavaRDD<Rating> validation = ratings.filter(v -> v._1() >= 6 && v._1() < 8)
                .values()
                .repartition(numPartitions).cache();

        /**
         * 测试集合
         */
        JavaRDD<Rating> test = ratings.filter(v -> v._1() >= 8)
                .values()
                .cache();


        long numTraining = training.count();
        long numValidation = validation.count();
        long numTest = test.count();

        System.out.println("numTraining:" + numTraining + "numValidation:" + numValidation + "numTest:" + numTest);

        int[] ranks = new int[]{10, 11, 12};
        double[] lambdas = new double[]{0.01};
        int[] numIters = new int[]{8, 9, 10};

        MatrixFactorizationModel bestModel = null;
        double bestValidationRmse = Double.MAX_VALUE;
        int bestRank = 0;
        double bestLambda = -0.01;
        int bestNumIter = 0;

        for (int rank : ranks) {
            for (double lambda : lambdas) {
                for (int numIter : numIters) {
                    /**
                     * 每一次训练的模型
                     */
                    MatrixFactorizationModel model = ALS.train(training.rdd(), rank, numIter, lambda);

                    System.out.println(model);

                    /**
                     * 计算误差
                     */
                    Double validationRmse = computeRmse(model, validation, numValidation);

                    if (validationRmse < bestValidationRmse) {
                        bestValidationRmse = validationRmse;
                        bestModel = model;
                        bestRank = rank;
                        bestLambda = lambda;
                        bestNumIter = numIter;
                    }
                }
            }
        }

        double testRmse = computeRmse(bestModel, test, numValidation);

        System.out.println("测试数据集在最佳训练集合 rank = " + bestRank + ",lambda = "
                + bestLambda + ",numIter = " + bestNumIter + ",RMSE = " + testRmse);

        /**
         * 计算均值
         */
        Double meanRating = training.union(validation).mapToDouble(v -> v.rating()).mean();

        /**
         * 计算标准误差值
         */
        Double baselineRmse = Math.sqrt(test.map(v -> (meanRating - v.rating()) * (meanRating - v.rating())).reduce((v1, v2) -> (v1 + v2) / numTest));

        double improvement = (baselineRmse - testRmse) / baselineRmse * 100;

        System.out.println("最佳训练的准确率提升了：" + String.format("%.2f", improvement) + "%.");

        /**
         * 构建最佳训练模型
         */
        bestModel = ALS.train(ratings.values().rdd(), bestRank, bestNumIter, bestLambda);

        return bestModel;

    }

    /**
     * 校验集预测数据和实际数据之间得均方根误差
     *
     * @param model
     * @param data
     * @param n
     * @return
     */
    public static Double computeRmse(MatrixFactorizationModel model, JavaRDD<Rating> data, Long n) {
        JavaPairRDD<Integer, Integer> integerIntegerJavaPairRDD = data.mapToPair(v -> new Tuple2<>(v.user(), v.product()));
        integerIntegerJavaPairRDD.foreach(integerIntegerTuple2 -> System.out.println(integerIntegerTuple2._1 + ":" + integerIntegerTuple2._2));
        JavaRDD<Rating> predictions = model.predict(integerIntegerJavaPairRDD);
        System.out.println("predictions: " + predictions.count());
        JavaRDD<Tuple2<Double, Double>> predictionsAndRatings = predictions
                .mapToPair(v -> new Tuple2<>(new Tuple2<>(v.user(), v.product()), v.rating()))
                .join(data.mapToPair(v -> new Tuple2<>(new Tuple2<>(v.user(), v.product()), v.rating()))).values();
        System.out.println("predictionsAndRatings: " + predictionsAndRatings.count());
        predictionsAndRatings.foreach(doubleDoubleTuple2 -> System.out.println(doubleDoubleTuple2._1 + ":" + doubleDoubleTuple2._2));
        Double reduce = predictionsAndRatings.map(v -> (v._1 - v._2) * (v._1 - v._2))
                .reduce((v1, v2) -> (v1 + v2) / n);
        return Math.sqrt(reduce);
    }
}
